package com.guchenbo.gpark.gd

import com.guchenbo.gpark.GparkContext

/**
 * @author guchenbo
 * @date 2022/3/21
 */
abstract class GD[T](gc: GparkContext) {
  var prev: GD[_] = _

  def compute(): Seq[T]

  //  支持的操作，filter、map、flatMap、groupBy
  //  支持的操作，reduce、fold

  def filter(func: T => Boolean): GD[T] = {
    val f = (seq: Seq[T]) => seq.filter(func)
    new MapGD(this, f)
  }

  def map[U](func: T => U): GD[U] = {
    val f = (seq: Seq[T]) => seq.map(func)
    new MapGD(this, f)
  }

  def flagMap[U](func: T => Seq[U]): GD[U] = {
    val f = (seq: Seq[T]) => seq.flatMap(func)
    new MapGD(this, f)
  }

  //  groupBy，非KV对的也可以分组
  def groupBy[K](func: T => K): GD[(K, Seq[T])] = {
    val f = (seq: Seq[T]) => seq.groupBy(func).toSeq
    new MapGD(this, f)
  }

  //  ，非KV对的也可以
  def reduce(func: (T, T) => T): T = {
    gc.runJob(this).reduce(func)
  }

  def fold(zero: T)(func: (T, T) => T): T = {
    val f: Seq[T] => T = (seq: Seq[T]) => seq.fold(zero)(func)
    f(gc.runJob(this))
  }

  def collect(): Seq[T] = {
    gc.runJob(this)
  }

  def print(): Unit = {
    println(collect().mkString(","))
  }

  def context: GparkContext = gc

  //  [_]表示自动推导
  def this(prev: GD[_]) = {
    this(prev.context)
  }
}

object GD {
  //  把KV对的RDD的方法封装到PairGDFunction里
  implicit def gd2PairGDFunction[K, V](gd: GD[(K, V)]): PairGDFunction[K, V] = {
    new PairGDFunction[K, V](gd)
  }

}

class DataGD[T](val seq: Seq[T], gc: GparkContext) extends GD[T](gc) {
  override def compute(): Seq[T] = seq
}
